package org.example.stream.consumer;

import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.example.stream.param.MessageBody;
import org.springframework.messaging.Message;

import static org.example.stream.producer.StreamProducer.*;

/**
 * 消费者
 * @author lly
 */
@Slf4j
public class AbstractStreamConsumer {

    public static <T> T getMessage(Message<MessageBody> msg, Class<T> type) {
        String messageId = msg.getHeaders().get(ROCKET_MQ_MESSAGE_ID,String.class);
        String tag = msg.getHeaders().get(ROCKET_TAGS,String.class);
        String topic = msg.getHeaders().get(ROCKET_MQ_TOPIC, String.class);
        Object object = JSON.parseObject(msg.getPayload().getBody(), type);
        log.info("Received Message topic:{} messageId:{},object:{}，tag:{}", topic, messageId, JSON.toJSONString(object), tag);
        return (T) object;
    }
}
